package com.aotain.coeus.unregistered.spark;

import java.io.IOException;
import java.net.URI;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

import com.aotain.coeus.unregistered.IPSearch2;
import com.aotain.coeus.unregistered.bean.IPGroup;
import com.aotain.coeus.unregistered.dao.RecordIntfDao;
import com.aotain.common.CommonFunction;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearch;
import com.hadoop.mapreduce.LzoTextInputFormat;

/**
 * SDS_RC_RECORDLIST表入库
 * @ClassName: SdsIpTiSpark
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年5月13日 上午9:47:05 
 *
 */
public class SdsRcRecordlistHive2 {

	public static void main(String[] args) throws IOException {

		if (args.length != 3){
			System.err.printf("Usage: <Input><Date><Output>");
			System.exit(1);
		}      

		String inpath = args[0];
		final String date = args[1].trim();
		String outpath = args[2];
		
//		Date dateNow = new Date();
//		Calendar cal = Calendar.getInstance();
//		cal.setTime(dateNow);
//		cal.add(Calendar.DAY_OF_MONTH, -1);
//		Date dateBef = cal.getTime();
//
//		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
//		final String date = String.format("%s%s0000", sdf.format(dateBef),hour);

		SparkConf conf = new SparkConf()
		.setAppName("SdsRcRecordlistHive2");
//		.set("spark.shuffle.consolidateFiles", "true");
		JavaSparkContext ctx = new JavaSparkContext(conf);
		
//		String DBJSON="{\"driverClassName\":\"com.mysql.jdbc.Driver\","
//				+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
//				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
//				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.97:3306/SDS\","
//				+ "\"userName\":\"sds\",\"validationQuery\":\"select sysdate()\"}";
		
		String DBJSON="{\"driverClassName\":\"oracle.jdbc.driver.OracleDriver\","
		+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
		+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
		+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:oracle:thin:@192.168.50.155:1521:bsmp\","
		+ "\"userName\":\"sds\",\"validationQuery\":\"select sysdate from dual\"}";
		
//		ApolloConfig ap = new ApolloConfig();
		final IPSearch ipsearch = new IPSearch(DBJSON);
		final IPSearch2 ipsearch2 = new IPSearch2(DBJSON);
		final Map<String,String> statusMap = RecordIntfDao.getInstance().getAllStatus();

		Configuration config = new Configuration();
		config.addResource("/etc/hadoop/conf");
		try {
			FileSystem fsTarget = FileSystem.get(URI.create(outpath),config);
			Path pathTarget = new Path(outpath);
			if(fsTarget.exists(pathTarget))
			{
				fsTarget.delete(pathTarget, true);
				System.out.println("Delete path " + outpath);
			}
		} catch (IllegalArgumentException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		//		config.set("fs.defaultFS", "hdfs://nameservice1");
		JavaPairRDD<LongWritable,Text> line = ctx.newAPIHadoopFile(inpath, LzoTextInputFormat.class, LongWritable.class, Text.class,config);
		//		JavaPairRDD<LongWritable,Text> line = ctx.newAPIHadoopFile(inpath, TextInputFormat.class, LongWritable.class, Text.class,config);
		JavaPairRDD<String, String> ss = line.mapToPair(new PairFunction<Tuple2<LongWritable,Text>, String,String>() {

			private static final long serialVersionUID = 6584386367142393259L;

			String tuple = "";
			String dip = "";
			String cityName = "";
			String groupID = "";
			String ipsegID = "";
			String status ;
			String destAreaCountry ;
			String destAreaCity;
			
			@Override
			public Tuple2<String, String> call(Tuple2<LongWritable, Text> v1)
					throws Exception {
				// TODO Auto-generated method stub
				String str = v1._2.toString();
				String[] splits = str.split("\\|", -1);
				if(splits.length > 7) {
					if(splits[6] != null && splits[7] != null && splits[2] != null && !"".equals(splits[6]) && !"".equals(splits[7]) && !"".equals(splits[2])) {
						String domainname = splits[6];
						String [] strArr = domainname.split("\\.",-1);
						String rootDomain = "";
						if(IsSpecialRoot(domainname) && strArr.length >= 3)
						{//特殊处理例如 .com.cn  .com.hk
							rootDomain = String.format("%s.%s.%s", strArr[strArr.length-3],
									strArr[strArr.length-2],strArr[strArr.length-1]);
						}
						else if(IsSpecialRoot2Level(domainname) && strArr.length>1)
						{
							//						 System.err.println("##########################" + domainname);
							rootDomain = String.format("%s.%s", 
									strArr[strArr.length-2],strArr[strArr.length-1]);
						}
						if(!"".equals(rootDomain)) {
							if(rootDomain.contains(":")) {
								rootDomain = rootDomain.substring(0,rootDomain.lastIndexOf(":"));
							}
							status = statusMap.get(rootDomain);
							if(status == null) {
								status = "0";
							}
							dip = splits[2];
							IPDatabase DestArea = ipsearch.getStartIP(dip);
							IPGroup ipGroup = ipsearch2.getStartIP(dip);
							if(DestArea != null) {
								destAreaCountry = DestArea.getCountryName();
								destAreaCity = DestArea.getCityName();
								//						sourceAreaProvince = SourceArea.getProviceName();
								if(destAreaCity != null) {
									cityName  = destAreaCity;
								} else {
									cityName = destAreaCountry;
								}
							}
							if(ipGroup != null) {
								 Long endIp = ipGroup.getEndIP();
								 Long dipL = CommonFunction.ip2int(dip);
								 if(dipL<endIp) {
									 groupID = ipGroup.getGroupID();
									 ipsegID = ipGroup.getID();
									 tuple = String.format("%s,%s,%s,%s,%s,%s,%s,", dip , cityName , rootDomain,date,status,groupID,ipsegID);
									 return new Tuple2<String,String>(tuple,"");
								 }
							}
						}
					}
				}
				return null;
			}

		}).filter(new Function<Tuple2<String,String>,Boolean>(){

			@Override
			public Boolean call(Tuple2<String, String> t) throws Exception {
				// TODO Auto-generated method stub
				if(t == null) {
					return false;
				} else {
					return true;
				}
			}

		}).distinct();

		ss.saveAsHadoopFile(outpath, Text.class, Text.class, TextOutputFormat.class);
		ctx.stop();
	}


	private static boolean IsSpecialRoot(String domain)
	{
		boolean blReturn = false;
		if(domain.contains(".com.")
				|| domain.contains(".co.")
				|| domain.contains("gd.cn")
				|| domain.contains(".cn.com")
				|| domain.contains(".gov.")
				|| domain.contains(".net.")
				|| domain.contains(".edu.")
				|| domain.contains(".org.")
				|| domain.contains(".ac."))
			blReturn = true;
		return blReturn;
	}

	private static boolean IsSpecialRoot2Level(String domain)
	{
		boolean blReturn = false;
		if(domain.contains(".com")
				|| domain.contains("cn")
				|| domain.contains(".net")
				|| domain.contains(".org"))
			blReturn = true;
		return blReturn;
	}
}
